package com.ibm.cps.processors;

import java.util.Collection;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.SchemaFormat;
import com.ibm.factories.MessageParserFactory;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.interfaces.AbstractMessagePublisher;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.PUBLISH_TYPE,Type=Info.ProcessorType.Other)
public class OutputProcessor extends BasicProcessor {

	private ITimeSeriesSchema parentSchema;
	private AbstractMessagePublisher publisher;
	private boolean isBuffer;
	
	@ValueParam @IgnoredInFunctionCall
	private String metadataString;

	private IMessageParser parser;

	public OutputProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
		this.parser = MessageParserFactory.createMessageParser(SchemaFormat.json, parentSchema);

	}
	
	@Override
	public void open() throws CPSException {
		publisher = PublishProcessorFactory.createOutputProcessor(metadataString);
	}
	
	@Override
	public void startBatch() {
		isBuffer = true;
	}

	@Override
	public void end() {
		isBuffer = false;
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message) throws CPSException {
		publisher.publishMessage(message, parser, isBuffer);
		return null;
	}

	@Override
	public void dispose() {
		publisher.close();
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() {
		return null;
	}

}
